package udpserver

import (
	"errors"
	"go-net/internal/udpserver/packet"
	"go-net/internal/udpserver/packet/deadline"
	"net"
	"sync"
	"sync/atomic"
)

var (
	ErrClosedListener      = errors.New("udp: listener closed")
	ErrListenQueueExceeded = errors.New("udp: listen queue exceeded")
)

type listener struct {
	udpConn        *net.UDPConn           // Udp服务连接
	accepting      atomic.Value           // bool 是否开启新接连监听
	connections    map[string]*Connection // 连接管理
	connLock       sync.Mutex             // 获取连接互斥锁
	acceptCh       chan *net.Conn
	acceptFilter   func([]byte) bool
	doneCh         chan struct{}
	connWG         sync.WaitGroup //  连接管理管道
	readWG         sync.WaitGroup
	readBufferPool *sync.Pool // 数据读取Buffer池
}

// Accept 接入新连接
func (l *listener) Accept() (*net.Conn, error) {
	select {
	case c := <-l.acceptCh:
		l.connWG.Add(1)
		return c, nil
	case <-l.doneCh:
		return nil, ErrClosedListener
	}
}

// Addr 返回监听的网络地址
func (l *listener) Addr() net.Addr {
	return l.udpConn.LocalAddr()
}

// Run 运行，循环读取数据
func (l *listener) Run() {
	defer l.readWG.Done()

	for {
		buf := *(l.readBufferPool.Get().(*[]byte))
		n, rAddr, err := l.udpConn.ReadFrom(buf)
		if err != nil {
			return
		}
		conn, ok, err := l.getConnection(rAddr, buf[:n])
		if err != nil {
			continue
		}
		if ok {
			_, _ = conn.buffer.Write(buf[:n])
		}
	}
}

// getConnection 使用连接地址和数据包，生成一个新的UDP连接对象
func (l *listener) getConnection(rAddr net.Addr, buf []byte) (*Connection, bool, error) {
	l.connLock.Lock()
	defer l.connLock.Unlock()
	conn, ok := l.connections[rAddr.String()]
	if !ok {
		if !l.accepting.Load().(bool) {
			return nil, false, ErrClosedListener
		}
		// 根据数据包来确定是否添加新的连接
		if l.acceptFilter != nil {
			if !l.acceptFilter(buf) {
				return nil, false, nil
			}
		}
		conn = l.newConnection(rAddr)

	}
	return conn, true, nil
}

// newConnection 使用传入的地址创建新的UDP连接
func (l *listener) newConnection(rAddr net.Addr) *Connection {
	return &Connection{
		listener:      l,
		rAddr:         rAddr,
		buffer:        packet.NewBuffer(),
		doneCh:        make(chan struct{}),
		writeDeadline: deadline.New(),
	}
}

// Close 关闭监听器
func (l *listener) Close() {
	l.accepting.Store(false)
}
